Video Thumbnail
2:47
2:14
clock icon Created with Sketch. 2 minutes

Solution: Unit Tests (Basic)

Although in the video I’m using Pydantic V1, we’ve updated the code example to Pydantic V2. As a result, there might be minor differences between the code you see in the video and the code in the Git repository. Notably, .dict() has been replaced by .model_dump().


Karol Borkowski

Where can I find the code for this challenge? I have access to the repo, but there is only the final code. The previous series has links to download the code in each lesson, but this one doesn't. Is that on purpose?

REPLY
Arjan Egges

Hi Karol, the idea is that you write the contents of the test file. I've clarified the text in the exercise and added a test_api_before.py file that you can use as a starting point.

REPLY
Philipp Walter

Hi all,
for those who used the extensive approach on day 22 like in the HotelAPI like me:
https://github.com/ArjanCodes/software-designer-mindset-complete-extension/blob/main/case_study/src_v2/hotel/db/db_interface.py
(access only for those who booked the course, thats why I only post a very small part of my solution - @Arjan&Team if this is not okay I delete this post asap)

I used dependency injection and changed the DBInterface as follows:

DataObject = dict[str, Any]

class DBInterface:
def __init__(self, db_class: type[Base], db_session: Session):
self.db_class = db_class
self.db_session = db_session

def read_by_id(self, id: int) -> DataObject:
data: Base | None = self.db_session.query(self.db_class).get(id)
if data is None:
return {}
return to_dict(data)
[...]

With this, init_db() is replaced by the context manager get_db() and provided as a FastAPI dependency in the single api_calls:

@app.get("/events/{event_id}")
async def api_get_event(event_id: int, database: Session = Depends(get_db)) -> EventResult:
db_interface = DBInterface(DBEvent, database)
return get_event(event_id, db_interface)

Would this be a common solution for larger projects (of course not the Day22 and 23 example)?

Thanks in advance.

REPLY
Andreas [ArjanCodes Team]

Hi!

Using the FastApi depends function is a very good solution for initializing db_session.
The solution of having a DBInterface could be a good way to reduce the amount of code, for example having basic functionality like CRUD for resources.
I like this approach! However, I can imagine that problems could arise if more specific functionality is needed. Which in a sense, can cause the DBInterface to be bloated and have too much responsibility.

REPLY
Manuel Escalona

Has someone experienced this error?

ImportError: cannot import name 'DeclarativeBase' from 'sqlalchemy.orm'

REPLY
Andreas [ArjanCodes Team]

What version of sqlalchemy are you using?

From sqlalchemy website:

Changed in version 2.0: The DeclarativeBase superclass supersedes the use of the declarative_base() function and registry.generate_base() methods; the superclass approach integrates with PEP 484 tools without the use of plugins. See ORM Declarative Models for migration notes.

REPLY
Manuel Escalona

Hi Andreas,

I had to uninstall and install SQLalchemy to sort the error.

Thanks

REPLY
Andreas [ArjanCodes Team]

Perfect, I am glad to hear that it worked out!

REPLY
Eduardo Motta de Moraes

Unless I'm mistaken, the way things are set up in the solution, the same database is reused for all the tests, which I believe isn't ideal, as anything you do in one test will impact all subsequent tests that also access the database. The results would change depending on the order in which the tests are run.

To solve this I created a pytest fixture that drops all the tables and creates them again for every test:


@pytest.fixture()
def test_db():
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)

I found this solution in this stackoverflow post. The post also suggest an arguably better (but slightly more complicated) approach of never commiting anything to the database by using nested transactions and rollbacks.

REPLY
Arjan Egges

Nice fix! When I have time, I'll add this to the solution for this challenge.

REPLY
Abhishek Roy

This change has been made in the repo now. However the fixture is never explicitly requested by another fixture or a test. I tried publishing a fix branch but I don't have permission. Currently the db is created once on line 16 of the solution, but we want to use the test_db fixture. To do this we need to delete line 16, and add test_db as an argument to the test_create_function() as such:

`def test_create_event(test_db) -> None:`

REPLY
Andreas [ArjanCodes Team]

Thanks for notifying us, we will look into it!

REPLY
Michael Brittain

I'm wondering how the delete event endpoint would be tested. In this case (for it to be a proper unit test) we'd need to manually insert the event into the database first before calling the delete endpoint, although I couldn't figure out how to do that.

REPLY
Oscar Bell

Hi Michael,

There are several ways to accomplish this. One is to use the create_event route first which will add an event to the database and within the same test case you can then delete the newly created event. Problem with this scenario is that you need to rely on the route create_event not to fail.

Solutions that are independent on the create_event route are shown below.
The 3 different test cases each deal with setting up the test environment differently.
1. Delete a previous inserted event with id 1 if it exists and insert a new event with id is 1;
2. Just adds another event into the database, similar to what create_event does but in this case it's not using the route but directly manipulating the database;
3. Delete all existing events and insert a new one.

def test_delete_event1() -> None:

db = next(override_get_db())
event = db.get(Event, 1)
if event:
print("Deleting previous event with event_id 1")
db.delete(event)
db.flush()

db_event = Event(title="Python Conference 2023",
id=1,
location="Amsterdam",
start_date=datetime.strptime("2023-03-15 09:00:00", "%Y-%m-%d %H:%M:%S"),
end_date=datetime.strptime("2023-03-18 16:00:00", "%Y-%m-%d %H:%M:%S"),
available_tickets=50)
db.add(db_event)
db.commit()
db.refresh(db_event)

test_client = TestClient(app=app)
response = test_client.delete("/events/1")
assert response.status_code == 200
assert response.json()['id'] == 1

def test_delete_event2() -> None:

db = next(override_get_db())

db_event = Event(title="Python Conference 2023",
location="Amsterdam",
start_date=datetime.strptime("2023-03-15 09:00:00", "%Y-%m-%d %H:%M:%S"),
end_date=datetime.strptime("2023-03-18 16:00:00", "%Y-%m-%d %H:%M:%S"),
available_tickets=50)
db.add(db_event)
db.commit()
db.refresh(db_event)
event_id = db_event.id

test_client = TestClient(app=app)
response = test_client.delete(f"/events/{event_id}")
assert response.status_code == 200
assert event_id == response.json()['id']

def test_delete_event3() -> None:

db = next(override_get_db())

stmt = delete(Event)
db.execute(stmt)

db_event = Event(title="Python Conference 2023",
location="Amsterdam",
start_date=datetime.strptime("2023-03-15 09:00:00", "%Y-%m-%d %H:%M:%S"),
end_date=datetime.strptime("2023-03-18 16:00:00", "%Y-%m-%d %H:%M:%S"),
available_tickets=50)
db.add(db_event)
db.commit()
db.refresh(db_event)
event_id = db_event.id

test_client = TestClient(app=app)
response = test_client.delete(f"/events/{event_id}")
assert response.status_code == 200
assert event_id == response.json()['id']

REPLY